package com.cy.tempflinkcdc.datachange;
import java.util.Calendar;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSONObject;
import com.cy.tempflinkcdc.service.TableDataConvertService;
import com.cy.tempflinkcdc.util.ApplicationContextUtil;
import com.cy.tempflinkcdc.util.FTPUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

import static com.cy.tempflinkcdc.util.FTPUtils.doPostFile;


/**
 * @author zhangzhongzhen wrote on 2024/3/24
 * @version 1.0
 * @description: 捕获数据变化后 执行的工具
 */
@Component
@Slf4j
public class DataChangeSink extends RichSinkFunction<DataChangeInfo> {

    public static String nowTime;


    @Value("${ftp.server}")
    private String server;

    @Value("${ftp.port}")
    private int port;

    @Value("${ftp.username}")
    private String username;

    @Value("${ftp.password}")
    private String password;

    @Value("${sql.path}")
    private String logfile;

    @Value("${delete.local-sql-file}")
    private Boolean deleteLocalSqlFileFlag = false;

    transient TableDataConvertService tableDataConvertService;

    @Override
    public void invoke(DataChangeInfo value, Context context) {
        // log.info("收到变更原始数据:{}", value);
        String sql = tableDataConvertService.convertSqlByDataChangeInfo(value);
        log.info("SQL:{}", sql);
        // logSql(sql); //记录sql语句到sql文件
        nowTime = writethefileintoSQL(nowTime, sql, logfile); // 记录sql语句到sql文件

    }

    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmm");
    private static BufferedWriter writer;





    public String writethefileintoSQL(String lasttime,String sql,String logfile) {

        long currentTime = getRoundedTime();
        String nowtime = dateFormat.format(new Date(currentTime));

        if (!nowtime.equals(lasttime) && StrUtil.isNotEmpty(lasttime)) {
            //发送一次包
            String alreadyFile = logfile + lasttime + ".sql";
            doPostFile(alreadyFile, server, port, username, password);
            log.info("postFTP", alreadyFile);
            if (deleteLocalSqlFileFlag) {
                FileUtil.del(logfile); // 配置文件中开启开关，是否删除文件
            }

        }
        log.info("sql:", sql);
        String newFileName = logfile + nowtime + ".sql";

        try {

            log.info("文件路径：{}",newFileName);
            writer = new BufferedWriter(new FileWriter(newFileName, true));
            writer.write(sql);
            writer.newLine();
            writer.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return nowtime;
    }


    private static long getRoundedTime() {
        long currentTime = System.currentTimeMillis();
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(currentTime);

        // 获取当前的分钟数
        int minutes = calendar.get(Calendar.MINUTE);

        // 将分钟数向下舍入到最近的10的倍数
        int roundedMinutes = (minutes / 10) * 10;
        // int roundedMinutes = minutes;

        // 设置舍入后的分钟数
        calendar.set(Calendar.MINUTE, roundedMinutes);
        calendar.set(Calendar.SECOND, 0);
        calendar.set(Calendar.MILLISECOND, 0);

        // 获取舍入后的时间
        long roundedTime = calendar.getTimeInMillis();
        return roundedTime;
    }






    /**
     * 在启动SpringBoot项目是加载了Spring容器，其他地方可以使用@Autowired获取Spring容器中的类；但是Flink启动的项目中，
     * 默认启动了多线程执行相关代码，导致在其他线程无法获取Spring容器，只有在Spring所在的线程才能使用@Autowired，
     * 故在Flink自定义的Sink的open()方法中初始化Spring容器
     */
    @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // this.rabbitTemplate = ApplicationContextUtil.getBean(RabbitTemplate.class);
            // this.confirmService = ApplicationContextUtil.getBean(ConfirmService.class);
            this.tableDataConvertService = ApplicationContextUtil.getBean(TableDataConvertService.class);
    }

}
